

#include "TcpServer.hpp"
#include "Acceptor.hpp"
#include "EventLoop.hpp"
#include "EventLoopThreadPool.hpp"
#include "Logger.hpp"
#include "TcpConnection.hpp"
#include <cstring>

namespace zephyr {

static EventLoop *CheckLoopNotNull(EventLoop *loop) {
    if (loop == nullptr) {
        LOG_FATAL("%s:%s:%d, mainLoop is null!!\n", __FILE__, __FUNCTION__, __LINE__);
    }
    return loop;
}

TcpServer::TcpServer(EventLoop *loop,
                     const InetAddress &listenaddr,
                     const std::string &name, Option option)
    : m_loop(CheckLoopNotNull(loop))
    , m_ipPort(listenaddr.toIpPort())
    , m_acceptor(new Acceptor(loop, listenaddr, ReusePort))
    , m_threadpool(new EventLoopThreadPool(loop, m_name))
    , m_connectionCallback()
    , m_messageCallback()
    , m_nextConnid(1)
    , m_started(0) {

    // 当有新连接时 acceptor 执行handRead 调用newConnectionCallback
    m_acceptor->setNewConnectionCallback(
        std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2)
    );
}

TcpServer::~TcpServer() {
    for (auto &item: m_connections) {
        TcpConnectionPtr conn(item.second);
        item.second.reset();
        conn->getLoop()->runInLoop(
            std::bind(&TcpConnection::connectDestroyed, conn)
        );
    }
}

void TcpServer::setThreadNum(int numThreads) {
    m_threadpool->setThreadNum(numThreads);
}

// 开启服务器监听
void TcpServer::start() {
    if (m_started.fetch_add(1) == 0) { // 防止一个server被多次启动
        m_threadpool->start(m_threadInitCallback);
        m_loop->runInLoop(std::bind(&Acceptor::listen, m_acceptor.get()));
    }
}

// EventLoop::loop --> Channel::handleEvent --> Acceptor::handleRead --> TcpServer::newConnection
void TcpServer::newConnection(int sockfd, const InetAddress &peeraddr) {
    EventLoop *ioloop = m_threadpool->getNextLoop();
    char buf[64] = {0};
    snprintf(buf, sizeof(buf), "-%s#%d", m_ipPort.c_str(), m_nextConnid);
    m_nextConnid ++;
    std::string connName = m_name + buf;
    LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s\n", m_name.c_str(), connName.c_str(), peeraddr.toIpPort().c_str());

    sockaddr_in local;
    ::memset(&local, 0, sizeof(local));
    socklen_t addrlen = sizeof(local);
    if (::getsockname(sockfd, (sockaddr *)&local, &addrlen) < 0) {
        LOG_ERROR("sockets::getLocalAddr");
    }

    InetAddress localAddr(local);
    TcpConnectionPtr conn(new TcpConnection(ioloop, connName, sockfd, localAddr, peeraddr));
    m_connections[connName] = conn;

    conn->setConnectionCallbakc(m_connectionCallback);
    conn->setMessageCallback(m_messageCallback);
    conn->setWriteCompleteCallback(m_writeCompletCallback);
    // 设置关闭连接的回调
    conn->setCloseCallback(std::bind(&TcpServer::removeConnection, this, std::placeholders::_1));
    ioloop->runInLoop(
        std::bind(&TcpConnection::connectEstablished, conn)
    );
}

void TcpServer::removeConnection(const TcpConnectionPtr &conn) {
    m_loop->runInLoop(std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}

void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn) {
    LOG_INFO("TcpSErver::removeConnectionInloop [%s] - connection %s\n", m_name.c_str(), conn->name().c_str());
    m_connections.erase(conn->name());
    EventLoop *loop = conn->getLoop();
    loop->queueInLoop(std::bind(&TcpConnection::connectDestroyed, conn));
}

}